package cn.zhang.violet.DataLabel

import cn.zhang.violet.config.ConfigHandler
import cn.zhang.violet.utils.{Jpools, TagUtils}
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable


object DataToLabel {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf()
    sparkConf.setAppName("数据标签化")
    sparkConf.setMaster("local[*]")
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)
    val sQLContext = new SQLContext(sc)
    //广播数据。得到app名称
    // 读取字典数据中的数据
    val parquetData = sQLContext.read.parquet(ConfigHandler.parquetFilePath)
      .filter(TagUtils.userIfFilterCondition)

    //对parquet数据进行mapPartition
    parquetData.mapPartitions(iter => {
      val tagMap = new mutable.HashMap[String, Map[String, Int]]()

      val jedis = Jpools.getJedis

      iter.foreach(row => {

        var currentLineMap = Map[String, Int]()

        // 获取用户id  放在最前面作为用户标识
        val userId = TagUtils.datainUserOneId(row)

        // 提取关键字段
        val adSpaceType = row.getAs[Int]("adspacetype")  //广告位
        val adSpaceTypeName = row.getAs[String]("adspacetypename") //广告位名称

        // 提取关键字段
        val appId = row.getAs[String]("appid")  //appid
        val appName = row.getAs[String]("appname") //appname

        // 渠道
        val channelId = row.getAs[Int]("adplatformproviderid")
        val client = row.getAs[Int]("client")  //操作系统
        val network = row.getAs[String]("networkmannername") //联网方式
        val ispName = row.getAs[String]("ispname")
        val keywords = row.getAs[String]("keywords")

        val pName = row.getAs[String]("provincename")
        val cName = row.getAs[String]("cityname")


        // 广告位类型
        if (adSpaceType > 9) currentLineMap += ("LC"+adSpaceType -> 1) else if (adSpaceType > 0) currentLineMap +=(("LC0"+adSpaceType, 1))
       //广告位名称
        if (StringUtils.isNotEmpty(adSpaceTypeName)) currentLineMap += ("LN"+ adSpaceTypeName ->1)

        //appname
        if (StringUtils.isEmpty(appName)) {
          if (StringUtils.isNotEmpty(appId)) {
            val appId2Name = jedis.hget("appdict", appId)
            if (StringUtils.isNotEmpty(appId2Name)) currentLineMap += ("APP"+appId2Name -> 1)
          }
        } else currentLineMap += ("APP"+appName -> 1)

        //运营商
        if (channelId > 0) currentLineMap += ("CN"+channelId -> 1)
        client match {
          case 1 => currentLineMap += (("D00010001", 1))
          case 2 => currentLineMap += (("D00010002", 1))
          case 3 => currentLineMap += (("D00010003", 1))
          case _ =>  //这种的就不要了
        }
         //联网
        network.toUpperCase match  {
          case "WIFI" => currentLineMap += (("D00020001", 1))
          case "4G" => currentLineMap += (("D00020002", 1))
          case "3G" => currentLineMap += (("D00020003", 1))
          case "2G" => currentLineMap += (("D00020004", 1))
          case _ =>
        }


        ispName.toUpperCase match  {
          case "移动" => currentLineMap += (("D00030001", 1))
          case "联通" => currentLineMap += (("D00030002", 1))
          case "电信" => currentLineMap += (("D00030003", 1))
          case _ =>
        }
        //关键字
        keywords.split("\\|")
          .filter(kw => kw.size >= 3 && kw.size <= 8)
          .foreach(word => currentLineMap += (("K"+word, 1)))

        if (StringUtils.isNotEmpty(pName)) currentLineMap += (("ZP"+pName, 1))
        if (StringUtils.isNotEmpty(cName)) currentLineMap += (("ZC"+cName, 1))

        tagMap.put(userId, currentLineMap)
      })

      jedis.close()
      tagMap.toIterator
    }).reduceByKey{
      (map1,map2) => (map1.toList ++ map2.toList).groupBy(_._1).mapValues(list => list.map(_._2).sum)
    }.saveAsTextFile("G:\\data\\log\\Tag")
    sc.stop()
  }
}
